package com.bbx.flink.table_api_or_sql;
import com.bbx.flink.demo.entity.UserAction;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import java.time.Duration;

import static org.apache.flink.table.api.Expressions.$;

/**
 * Flink table api   eventTime   watermark demo
 * 通过table  api 创建 event time watermark
 */
@Slf4j
public class TimeAttributeDataStreamEventTimeDemo {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //从Flink1.12 开始  默认的处理时间为   EventTime
//        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //水印策略--
        WatermarkStrategy<UserAction> watermarkStrategy = WatermarkStrategy
                .<UserAction>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                .withTimestampAssigner((userAction, l) -> userAction.getActive_time());

        SingleOutputStreamOperator<UserAction> dataStream = env.socketTextStream("192.168.10.131", 10003)
                .map(i -> {
                    log.info("~~~~~~~~~[{}]", i);
                    String[] split = i.split(",");
                    UserAction userAction = new UserAction(split[0], split[1], Long.parseLong(split[2]));
                    log.info("!!!!![{}]", userAction.toString());
                    return userAction;
                })
                .assignTimestampsAndWatermarks(watermarkStrategy);
        //此处的流必须是设置了watermark strategy 的，否侧就会报错     Recovery is suppressed by NoRestartBackoffTimeStrategy
        //当Stream 转换到Table 时有 两种方式来定义时间属性（Time attribute）  调用.rowTime() 的属性存在时，将会覆盖该属性的值，不存在时创建新的属性 ，以下创建table 的语句二选一
        Table table = tableEnv.fromDataStream(dataStream,$("id"), $("name") ,$("active_time") ,$("active_time_time").rowtime());  //  active_time_time  不存在，创建
//        Table table = tableEnv.fromDataStream(dataStream, $("id"),$("active_time").rowtime(),$("name")  );  //active_time  存在，被覆盖


        tableEnv.toAppendStream(table,Row.class).print();

        env.execute();

    }

}
